package org.apache.flink.runtime.resourcemanager.registration;

import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.resourceprovider.ResourceProviderMemoryConfiguration;
import org.apache.flink.runtime.resourceprovider.ResourceProviderGateway;
import org.apache.flink.util.Preconditions;

/** This class extends the {@link ResourceProviderConnection}, adding the worker information. */
public class WorkerRegistration<WorkerType extends ResourceIDRetrievable>
        extends ResourceProviderConnection {

    private final WorkerType worker;

    private final int jmxPort;

    private final HardwareDescription hardwareDescription;

    private final ResourceProviderMemoryConfiguration memoryConfiguration;

    private final ResourceProfile totalResourceProfile;

    private final ResourceProfile defaultSlotResourceProfile;

    private final String nodeId;

    public WorkerRegistration(
    		ResourceProviderGateway taskExecutorGateway,
            WorkerType worker,
            int jmxPort,
            HardwareDescription hardwareDescription,
            ResourceProviderMemoryConfiguration memoryConfiguration,
            ResourceProfile totalResourceProfile,
            ResourceProfile defaultSlotResourceProfile,
            String nodeId) {

        super(worker.getResourceID(), taskExecutorGateway);
        this.worker = Preconditions.checkNotNull(worker);
        this.jmxPort = jmxPort;
        this.hardwareDescription = Preconditions.checkNotNull(hardwareDescription);
        this.memoryConfiguration = Preconditions.checkNotNull(memoryConfiguration);
        this.totalResourceProfile = Preconditions.checkNotNull(totalResourceProfile);
        this.defaultSlotResourceProfile = Preconditions.checkNotNull(defaultSlotResourceProfile);
        this.nodeId = Preconditions.checkNotNull(nodeId);
    }

    public WorkerType getWorker() {
        return worker;
    }

    public int getJmxPort() {
        return jmxPort;
    }

    public HardwareDescription getHardwareDescription() {
        return hardwareDescription;
    }

    public ResourceProviderMemoryConfiguration getMemoryConfiguration() {
        return memoryConfiguration;
    }

    public ResourceProfile getDefaultSlotResourceProfile() {
        return defaultSlotResourceProfile;
    }

    public ResourceProfile getTotalResourceProfile() {
        return totalResourceProfile;
    }

    public String getNodeId() {
        return nodeId;
    }
}
